package com.sunday.common.reactor.e6_Filtering_Sequence;

import com.sunday.common.reactor.SubscribeUtils;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * A.4. 过滤序列
 */
@Slf4j
public class FilteringSequence {

    /**
     * 我想过滤一个序列
     */

    /**
     * 基于任意标准：（Flux|Mono#filter)
     * 通过使用 filter() 操作符，您可以过滤序列中的值。
     */
    @Test
    public void filter() {
        Flux<Integer> flux = Flux.range(0, 10)
                .filter(i -> i % 2 == 0);
        SubscribeUtils.subscribe(flux);
    }

    /**
     * 异步计算：（Flux|Mono#filterWhen)
     * 通过使用 filterWhen() 操作符，您可以过滤序列中的值，该值是通过异步计算确定的。
     */
    @Test
    public void filterWhen() {
        Flux<Integer> flux = Flux.range(0, 10)
                .filterWhen(i -> i % 2 == 0 ? Mono.just(true) : Mono.just(false));
        SubscribeUtils.subscribe(flux);
    }

    /**
     * 限制发出对象的类型：（Flux|Mono#ofType)
     * 通过使用 ofType() 操作符，您可以限制序列中发出的对象的类型。
     */
    @Test
    public void ofType() {
        Flux<Object> flux = Flux.just(1, "apple", 2.5, "orange", 3, "banana");
        SubscribeUtils.subscribe(flux
                // ofType 会过滤掉不是指定类型的元素
                .ofType(Integer.class));
    }

    /**
     * 通过完全忽略这些值：（Flux.ignoreElements（）|Mono.ignoreElement（）ignoreElements)
     * 通过使用 ignoreElements() 操作符，您可以完全忽略这些值，只发出完成或错误信号。
     */
    @Test
    public void ignoreElements() {
        Mono<Integer> flux = Flux.range(0, 10)
                // ignoreElements 会忽略所有元素，只发出完成或错误信号
                .ignoreElements();
        SubscribeUtils.subscribe(flux);
    }

    /**
     * 通过忽略重复项：
     * 在整个序列（逻辑集）中：Flux#distinct
     */
    @Test
    public void distinct() {
        Flux<Integer> flux = Flux.just(1, 2, 2, 3, 3, 3, 1, 1, 4, 4, 4)
                // distinct 会忽略重复元素
                .distinct();
        SubscribeUtils.subscribe(flux);
    }

    /**
     * 通过忽略重复项：
     * 在随后发出的项目之间（重复数据删除）：Flux#distinctUntilChanged
     */
    @Test
    public void distinctUntilChanged() {
        Flux<Integer> flux = Flux.just(1, 2, 2, 3, 3, 3, 1, 1, 4, 4, 4)
                // distinctUntilChanged 会忽略重复元素，但是只会忽略随后发出的重复元素
                .distinctUntilChanged();
        SubscribeUtils.subscribe(flux);
    }

    /**
     * 我只想保留序列的一个子集
     */

    /**
     * 通过取 N 个元素：
     * 在序列的开头：Flux#take（long）
     * 从上游请求无限量：Flux#take（long， false）
     * 基于持续时间：Flux#take（Duration）
     * 只有第一个元素，作为 Mono：Flux#next（）
     * 在序列的末尾：Flux#takeLast
     * 直到满足条件（含）：Flux#takeTil（基于谓词）、Flux#takeUntilOther（基于随播发布商）
     * 当满足条件时（独占）：Flux#takeWhile
     */
    @Test
    public void take() {
        Flux<String> flux = Flux.generate(
                // 我们提供初始状态值 0。
                () -> 0,
                // 我们提供一个生成器函数，它将状态值作为输入，并返回下一个值。
                (state, sink) -> {
                    log.info("3 x " + state + " = " + 3 * state);
                    // 我们使用 sink.next(T)方法来生成下一个值。
                    sink.next("3 x " + state + " = " + 3 * state);
                    if (state == 3) {
                        try {
                            TimeUnit.SECONDS.sleep(3);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    }
                    // 我们还使用它来选择何时停止。
                    if (state == 10) sink.complete();
                    // 我们返回一个在下一次调用中使用的新状态（除非 序列在此终止）
                    return state + 1;
                });

        log.info("take 会取前 n 个元素");
        // take 会取前 n 个元素
        SubscribeUtils.subscribe(flux.take(5));

        log.info("take 会取前 n 个元素，但是不会等待上游完成");
        // take 会取前 n 个元素，但是不会等待上游完成
        SubscribeUtils.subscribe(flux.take(5, false));

        log.info("take 会取前 n 个元素，但是不会等待上游完成");
        // take 会取前 n 个元素，但是不会等待上游完成
        SubscribeUtils.subscribe(flux.take(Duration.ofSeconds(1)));

        log.info("next 会取第一个元素，作为 Mono");
        // next 会取第一个元素，作为 Mono
        SubscribeUtils.subscribe(flux.next());

        log.info("takeLast 会取后 n 个元素");
        // takeLast 会取后 n 个元素
        SubscribeUtils.subscribe(flux.takeLast(5));

        log.info("takeUntil 当谓词返回 true 时，它会发射数据。当谓词返回 false 时，它会停止发射数据。");
        // takeUntil 当谓词返回 true 时，它会发射数据。当谓词返回 false 时，它会停止发射数据。
        SubscribeUtils.subscribe(flux.takeUntil(i -> i.equals("3 x 1 = 3")));

        log.info("takeUntilOther 会发出来自源序列的元素，直到另一个序列发出元素为止。");
        // takeUntilOther 会发出来自源序列的元素，直到另一个序列发出元素为止。
        Mono<Long> terminateWhen = Mono.delay(Duration.ofSeconds(1));
//        SubscribeUtils.subscribe(terminateWhen);
        SubscribeUtils.subscribe(flux.takeUntilOther(terminateWhen));

        log.info("takeWhile 当谓词返回 true 时，它会发射数据。当谓词返回 false 时，它会停止发射数据。");
        // 当谓词返回 true 时，它会发射数据。当谓词返回 false 时，它会停止发射数据。
        SubscribeUtils.subscribe(flux.takeWhile(i -> !i.equals("3 x 3 = 9")));

    }

    /**
     * 最多服用 1 个元素：
     * 在特定位置：Flux#elementAt
     * 最后：.takeLast（1）
     * 如果为空，则发出错误：Flux#last（）
     * 如果为空，则发出默认值：Flux#last（T）
     */
    @Test
    public void elementAt() {
        Flux<String> flux = Flux.just("apple", "orange", "banana");

        log.info("elementAt 会取指定位置的元素");
        // elementAt 会取指定位置的元素
        SubscribeUtils.subscribe(flux.elementAt(1));

        log.info("takeLast 会取最后一个元素");
        // takeLast 会取最后一个元素
        SubscribeUtils.subscribe(flux.takeLast(1));

        log.info("last 会取最后一个元素，如果为空，则发出错误");
        // last 会取最后一个元素，如果为空，则发出错误
        SubscribeUtils.subscribe(flux.last());

        log.info("last 会取最后一个元素，如果为空，则发出默认值");
        // last 会取最后一个元素，如果为空，则发出默认值
        SubscribeUtils.subscribe(Flux.empty().last("default"));
    }

    /**
     * 通过跳过元素：
     * 在序列的开头：Flux#skip（long）
     * 基于持续时间：Flux#skip（Duration）
     * 在序列的末尾：Flux#skipLast
     * 直到满足条件（含）：Flux#skipTil（基于谓词）、Flux#skipUntilOther（基于随附发布商）
     * 满足条件时（独占）：Flux#skipWhile
     */
    @Test
    public void skip() throws InterruptedException {
        Flux<String> flux = Flux.just("apple", "orange", "banana", "test");

        log.info("skip 会跳过前 n 个元素");
        // skip 会跳过前 n 个元素
        SubscribeUtils.subscribe(flux.skip(1));

        log.info("skip(Duration timespan) 根据时间跳过, 如果生成元素时间大于skip时间，则不会产生跳过情况");
        // skip(Duration timespan) 根据时间跳过, 如果生成元素时间大于skip时间，则不会产生跳过情况
        SubscribeUtils.subscribe(flux.delayElements(Duration.ofSeconds(1)).skip(Duration.ofSeconds(3)));
        TimeUnit.SECONDS.sleep(5);

        log.info("skipLast 会跳过后 n 个元素");
        // skipLast 会跳过后 n 个元素
        SubscribeUtils.subscribe(flux.skipLast(1));

        log.info("skipUntil 当谓词返回 true 时，它会跳过数据。当谓词返回 false 时，它会停止跳过数据。");
        // skipUntil 当谓词返回 true 时，它会跳过数据。当谓词返回 false 时，它会停止跳过数据。
        SubscribeUtils.subscribe(flux.skipUntil(i -> i.equals("banana")));

        log.info("skipUntilOther 会跳过来自源序列的元素，直到另一个序列发出元素为止。");
        // skipUntilOther 会跳过来自源序列的元素，直到另一个序列发出元素为止。
        Mono<Long> terminateWhen = Mono.delay(Duration.ofSeconds(3));
        SubscribeUtils.subscribe(flux.delayElements(Duration.ofSeconds(1)).skipUntilOther(terminateWhen));
        TimeUnit.SECONDS.sleep(5);

        log.info("skipWhile 当谓词返回 true 时，它会跳过数据。当谓词返回 false 时，它会停止跳过数据。");
        // skipWhile 当谓词返回 true 时，它
        SubscribeUtils.subscribe(flux.skipWhile(i -> !i.equals("banana")));
    }

    /**
     * 按抽样项目：
     * 按持续时间：Flux#sample（Duration）
     * 但将第一个元素保留在采样窗口中，而不是最后一个元素：sampleFirst
     * 通过基于发布者的窗口：Flux#sample（Publisher）
     * 基于发布者“超时”：Flux#sampleTimeout（每个元素触发一个发布者，如果该发布者不与下一个发布者重叠，则发出该元素）
     */
    @Test
    public void sample() throws InterruptedException {
        /**
         * 因此，flux.sample 在采样间隔内仅发出最近的元素，而 flux.sampleFirst 则会在每次采样时立即发出最新的元素。
         */
        Flux<Integer> flux = Flux.range(0, 8);

        log.info("sample(Duration period) 这个方法会根据指定的时间间隔对源 Flux 进行采样。它会返回在采样间隔内最近的元素。如果在采样间隔内没有元素，则不会发出任何值。");
        // sample(Duration period) 这个方法会根据指定的时间间隔对源 Flux 进行采样。它会返回在采样间隔内最近的元素。如果在采样间隔内没有元素，则不会发出任何值。
        SubscribeUtils.subscribe(flux.delayElements(Duration.ofSeconds(1)).sample(Duration.ofSeconds(2)));
        TimeUnit.SECONDS.sleep(10);

        log.info("sampleFirst(Duration period) 相反，flux.sampleFirst 方法会在每次采样时立即发出源 Flux 中的最新元素。这意味着，无论在采样间隔内是否有新元素，它都会立即发出最新的元素");
        // sampleFirst(Duration period) 相反，flux.sampleFirst 方法会在每次采样时立即发出源 Flux 中的最新元素。这意味着，无论在采样间隔内是否有新元素，它都会立即发出最新的元素
        SubscribeUtils.subscribe(flux.delayElements(Duration.ofSeconds(1)).sampleFirst(Duration.ofSeconds(2)));
        TimeUnit.SECONDS.sleep(10);

        log.info("sampleTimeout(Publisher sampler) 基于发布者“超时”");
        // sampleTimeout(Publisher sampler) 基于发布者“超时”
        SubscribeUtils.subscribe(flux.delayElements(Duration.ofSeconds(1)).sampleTimeout(i -> Mono.delay(Duration.ofSeconds(3))));
        TimeUnit.SECONDS.sleep(10);
    }

    /**
     * 我期望最多 1 个元素（如果不止一个，则出错）......
     * 如果序列为空，我想要一个错误：Flux#single（）
     * 如果序列为空，我想要一个默认值：Flux#single（T）
     * 我也接受一个空序列：Flux#singleOrEmpty
     */
    @Test
    public void single() {
        /**
         * Flux#single() 用于获取源 Flux 中的单个元素，要求源 Flux 正好只包含一个元素；而 Flux#single(T) 也用于获取源 Flux 中的单个元素，但允许提供默认值以处理多个元素或者空序列的情况。
         */
        Flux<String> flux = Flux.just("apple");

        log.info("Flux#single(): 这个方法返回源 Flux 中的单个元素。如果源 Flux 包含超过一个元素或者没有元素，则会抛出 IndexOutOfBoundsException 或者 NoSuchElementException 异常。这意味着，当你确定源 Flux 只包含一个元素时才应该使用此方法，否则会引发异常。");
        // Flux#single(): 这个方法返回源 Flux 中的单个元素。如果源 Flux 包含超过一个元素或者没有元素，则会抛出 IndexOutOfBoundsException 或者 NoSuchElementException 异常。这意味着，当你确定源 Flux 只包含一个元素时才应该使用此方法，否则会引发异常。
        SubscribeUtils.subscribe(flux.single());

        log.info("Flux#single(T): 这个方法接受一个默认值作为参数。它与 Flux#single() 类似，但在源 Flux 包含超过一个元素或者没有元素时，不会抛出异常，而是返回指定的默认值。");
        // Flux#single(T): 这个方法接受一个默认值作为参数。它与 Flux#single() 类似，但在源 Flux 包含超过一个元素时报错，没有元素时，不会抛出异常，而是返回指定的默认值。
        SubscribeUtils.subscribe(Flux.empty().single("default"));

        log.info("singleOrEmpty 会取第一个元素，如果不止一个，则出错");
        // singleOrEmpty 会取第一个元素，如果不止一个，则出错
        SubscribeUtils.subscribe(Flux.empty().singleOrEmpty());
    }

}
